Enable configuration of a CDC mutation info Callable for CDC Writes into BigQuery #32878
Conversation
prodriguezdefino
changed the title
prodriguezdefino
changed the title
|
Assigning reviewers. If you would like to opt out of this review, comment R: @shunping for label python. Available commands:
The PR bot will only process comments in the main thread (not review comments). |
|
Reminder, please take a look at this pr: @shunping @Abacn @johnjcasey |
|
Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment R: @jrmccluskey for label python. Available commands:
|
|
Reminder, please take a look at this pr: @jrmccluskey @damccorm @ahmedabu98 |
| @@ -2550,7 +2577,7 @@ def __init__( | |||
| use_at_least_once=False, | |||
| with_auto_sharding=False, | |||
| num_storage_api_streams=0, | |||
| use_cdc_writes: bool = False, | |||
| use_cdc_writes: UseCdcWrites = False, | |||
There was a problem hiding this comment.
Could you add some more context on why we need this PR? What use case is it enabling which previously was impossible/is there a reason having this as a parameter is better than just recommending users do preprocessing before the write step? As much as we can, I'd prefer to keep custom user code out of the IO for debugability.
I'll also note that this parameter is now very overloaded. I'd probably prefer an extra param here over repurposing this one.
There was a problem hiding this comment.
But lets start with the first question, because I'm not sure about the change as a whole yet
There was a problem hiding this comment.
Originally this change was part of #32529 and given the push to get it included for the v2.60.0 release @ahmedabu98 recommended partition it in two different PRs.
Before #32529 Python SDK was not capable of using CDC writes into BQ and this change brings parity with BigQueryIO by exposing a lambda to include Row Mutation information as part of the row to be ingested. Currently, a SDK user would need to know how to structure their Rows to be ingested using CDC or, if they use Dicts as their data format, their provided schema should include the row mutation information making it not matching with the actual BigQuery table schema they want to write to (otherwise the xlang protocol wouldn't work).
There was a problem hiding this comment.
I'll also note that this parameter is now very overloaded. I'd probably prefer an extra param here over repurposing this one.
We have similar overloads in this same file, see here and also we have other instances on where the overload is silently added on a typeless argument.
The strict type definition simplifies the understanding of the shape for the argument, and to some degree the validations made later (would have been even simpler and more strict with Python > 3.10).
There was a problem hiding this comment.
Currently, a SDK user would need to know how to structure their Rows to be ingested using CDC or, if they use Dicts as their data format, their provided schema should include the row mutation information making it not matching with the actual BigQuery table schema they want to write to (otherwise the xlang protocol wouldn't work).
Isn't this still true with your change? The only difference is that they're now encapsulating the conversion logic in their Write transform instead of having a step to do it beforehand. Basically, as I understand it, the following 2 blocks are equivalent (and not particularly different to write):
pcoll
| WriteToBigQuery(..., use_cdc_writes=my_fn, ...)
and:
pcoll
| beam.Map(my_fn)
| WriteToBigQuery(..., use_cdc_writes=True, ...)
the main difference is that the 2nd is easier to debug IMO. Am I right that those are equivalent in all cases or am I missing something?
this change brings parity with BigQueryIO
This is a good reason to consider the change, but not enough to add it IMO
We have similar overloads in this same file, see here and also we have other instances on where the overload is silently added on a typeless argument.
To be honest, I don't love that usage either. However, there are a few differences there:
- It is not a user facing API
tableis a more general arg thanuse_cdc_writes(which is no longer descriptive with this change)
|
Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment R: @liferoad for label python. Available commands:
|
This change enables the configuration of a Callable function to populate the CDC mutation information on a per record basis when using BigQuery as destination with CDC writes.
Given that
beam.io.WriteToBigQuerysupports input data asDictorbeam.Row, the configured CDC info Callable should conform to one of those 2 types:Callable[[Dict], Dict]orCallable[[beam.Row], beam.Row].Example configuration for Dict type data:
... data = [ # record: (name, value) { 'name': 'cdc_test', 'value': 5 }, { 'name': 'cdc_test', 'value': 3 } ] schema = { "fields": [{ "name": "name", "type": "STRING" }, { "name": "value", "type": "INTEGER" }] } def cdc_info_fn(data: Dict) -> Dict: return { 'mutation_type': 'UPSERT', 'change_sequence_number': 'AAA/' + str(data["value"]) } with beam.Pipeline(argv=self.args) as p: _ = ( p | beam.Create(data) | beam.io.WriteToBigQuery( table=table_id, method=beam.io.WriteToBigQuery.Method.STORAGE_WRITE_API, use_cdc_writes=cdc_info_fn, schema=schema, primary_key=["name"], use_at_least_once=True)) # expected data written on BQ table after pipeline finishes # { # 'name': 'cdc_test', 'value': 5 # }Analogously, in case of processing Beam Rows the pipeline would look like:
... row_data = [ beam.Row(name="cdc_test", value=5, route=3), beam.Row(name="cdc_test", value=3, route=3), beam.Row(name="cdc_test", value=2, route=1) ] def cdc_info_rows(row: beam.Row) -> beam.Row: return beam.Row( mutation_type="UPSERT", change_sequence_number="AAA/" + str(row.value + row.route)) with beam.Pipeline(argv=self.args) as p: _ = ( p | beam.Create(row_data) | beam.io.WriteToBigQuery( table=table_id, method=beam.io.WriteToBigQuery.Method.STORAGE_WRITE_API, use_cdc_writes=cdc_info_rows, primary_key=["name"], use_at_least_once=True)) # expected data written on BQ table after pipeline finishes # { # 'name': 'cdc_test', 'value': 5, 'route': 3 # }GitHub Actions Tests Status (on master branch)
See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.